package com.atguigu.sparkcore.day02.kv

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Author atguigu
 * Date 2020/10/28 14:30
 */
object CombineByKey {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
        val sc: SparkContext = new SparkContext(conf)
        val rdd1 = sc.makeRDD(List(("a", 1), ("b", 5), ("a", 2), ("a", 7), ("a", 5), ("b", 2)), 2)
        // 计算平均值   和/个数   ()
        val rdd2: RDD[(String, (Int, Int))] = rdd1.combineByKey(
            v => (v, 1),
            {
                case ((sum, count), v) =>
                    (sum + v, count + 1)
            },
            {
                case ((sum1, count1), (sum2, count2)) =>
                    (sum1 + sum2, count1 + count2)
            }
        )
        /*rdd2
            .map {
                case (word, (sum, count)) => (word, sum.toDouble / count)
            }
            .collect
            .foreach(println)*/
        
        val rdd3 = rdd2.mapValues {
            case (sum, count) => sum.toDouble / count
        }
        rdd3.collect.foreach(println)
        
        
        sc.stop()
    }
}
/*
combineByKey
    零值是动态生成
    分区内和分区间的聚合逻辑不一样
    
    combineByKeyWithClassTag(createCombiner, mergeValue, mergeCombiners, defaultPartitioner(self))
aggregateByKey
    有零值
    分区内和分区间的聚合逻辑不一样
    combineByKeyWithClassTag[U]((v: V) => cleanedSeqOp(createZero(), v),
      cleanedSeqOp, combOp, partitioner)
foldByKey
    有零值
    分区内和分区间的聚合逻辑一样
    combineByKeyWithClassTag[V]((v: V) => cleanedFunc(createZero(), v),
      cleanedFunc, cleanedFunc, partitioner)
reduceByKey
    没有零值
    分区内和分区间的聚合逻辑一样
    combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner)
    
都支持预聚合(map端聚合)
 */
